package hadoop.ch06.v17124080238;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.HBaseConfiguration;
import org.apache.hadoop.hbase.HColumnDescriptor;
import org.apache.hadoop.hbase.HTableDescriptor;
import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.client.*;
import org.apache.hadoop.hbase.util.Bytes;
import java.io.IOException;


public class HBaseDemo63 {
    static private Configuration conf = HBaseConfiguration.create();
    //主节点
    private static final String rootDir = "hdfs://master:9000/hbase";
    //从节点
    private static final String quorum = "master,slave0,slave1";
    //是否为分布式
    private static final String distributed = "true";
    //Hbase配置
    static {
        conf.set("hbase.rootdir",rootDir);
        conf.set("hbase.zookeeper.quorum",quorum);
        conf.set("hbase.cluster.distributed",distributed);
    }

    //创建表
    //tName：表名，family：列簇
    static public void createTable(String tName, String[] families) throws IOException {
        //创建数据库连接
        Connection con = ConnectionFactory.createConnection(conf);
        Admin admin = con.getAdmin();
        //选择表名
        TableName tableName = TableName.valueOf(tName);
        //查看表是否存在
        if(admin.tableExists(tableName)) {   //exception
            System.out.println("表已存在");
            admin.close();
            System.exit(0);
        }
        //加入描述
        HTableDescriptor hTableDescriptor = new HTableDescriptor(tName);
        //输入列簇信息
        for (String family : families) {
            HColumnDescriptor hColumnDescriptor = new HColumnDescriptor(family);
            hTableDescriptor.addFamily(hColumnDescriptor);
        }
        //正式创建表
        admin.createTable(hTableDescriptor);
        //关闭连接
        admin.close();
    }

    //插入一个数据
    static public void insert(String tableName
            , String rowKey, String family, String column, String value)
            throws IOException {
        //选择要操作的表
        HTable hTable = new HTable(conf, tableName);
        //构造数据
        Put put = new Put(Bytes.toBytes(rowKey)); //键值
        put.addColumn(
                Bytes.toBytes(family), //列簇
                Bytes.toBytes(column), //列
                Bytes.toBytes(value)   //值
        );
        //插入
        hTable.put(put);
        //关闭
        hTable.close();
    }

    //获取一个数据项
    static public void get(String tableName, String rowKey, String family, String column) throws IOException {
        //选择要操作的表
        HTable hTable = new HTable(conf, tableName);
        //通过get查询
        Get get = new Get(Bytes.toBytes(rowKey));
        Result record = hTable.get(get);
        String result = Bytes.toString(
                record.getValue(
                        Bytes.toBytes(family),  //列簇
                        Bytes.toBytes(column)  //列
                ));
        System.out.println(family + ":" + column + " " + result);
        hTable.close();
    }

    //扫描整个表
    static public void scan(String tableName) throws IOException {
        //选择要操作的表
        HTable hTable = new HTable(conf, tableName);
        Scan scan = new Scan();
        //执行查询
        ResultScanner rs = hTable.getScanner(scan);
        //输出
        for(Result r : rs) {
            String name = Bytes.toString(r.getRow());
            String age = Bytes.toString(
                    r.getValue( //输入列簇
                            Bytes.toBytes("info"),
                            Bytes.toBytes("age")
                    )
            );
            String id = Bytes.toString(
                    r.getValue(
                            Bytes.toBytes("member_id"),
                            Bytes.toBytes("id")
                    ));
            String birth = Bytes.toString(
                    r.getValue(
                            Bytes.toBytes("info"),
                            Bytes.toBytes("birthday")
                    ));
            String industry = Bytes.toString(
                    r.getValue(
                            Bytes.toBytes("info"),
                            Bytes.toBytes("industry")
                    ));
            String city = Bytes.toString(
                    r.getValue(
                            Bytes.toBytes("address"),
                            Bytes.toBytes("city")
                    ));
            String country = Bytes.toString(
                    r.getValue(
                            Bytes.toBytes("address"),
                            Bytes.toBytes("country")
                    ));
            //打印
            System.out.println(name + " " +
                    id + " "
                    + age + " " + birth + " " + industry +
                    " " + city + " " + country
            );
        }
        //关闭
        hTable.close();
    }

}
